#ifndef KAFKA_HPP
#define KAFKA_HPP

#include <unordered_map>
#include <mutex>
#include <atomic>
#include <string>
#include <memory>
#include <vector>
#include <thread>
#include <functional>

// 前向声明，避免包含librdkafka/rdkafkacpp.h
namespace RdKafka {
    class Producer;
    class KafkaConsumer;
    class Conf;
    class Topic;
    class Message;
}

typedef std::function<void(int, std::string)> kafka_handler;

class Kafka {
public:
    Kafka();
    ~Kafka();

    // 初始化Kafka连接
    bool init(const std::string& brokers);
    
    // 向Kafka指定的主题发布消息
    bool publish(const std::string& topic, int channel, const std::string& message);
    
    // 向Kafka订阅主题
    bool subscribe(const std::string& topic);
    
    // 取消订阅主题
    bool unsubscribe(const std::string& topic);
    
    // 开始消费消息
    void start_consuming();
    
    // 停止消费
    void stop();
    
    // 初始化业务层上报消息的回调对象
    void init_notify_handler(kafka_handler handler);

private:
    // 独立线程中接收订阅主题的消息
    void observe_topic_messages();

    // Kafka生产者和消费者实例
    RdKafka::Producer* producer_;
    RdKafka::KafkaConsumer* consumer_;
    
    // 生产者和消费者配置
    RdKafka::Conf* producer_conf_;
    RdKafka::Conf* consumer_conf_;
    
    // 主题缓存
    std::unordered_map<std::string, std::unique_ptr<RdKafka::Topic>> producer_topic_cache_;
    std::unordered_map<std::string, std::unique_ptr<RdKafka::Topic>> consumer_topic_cache_;
    
    // 主题订阅列表
    std::vector<std::string> subscribed_topics_;
    
    // 互斥锁
    std::mutex producer_mutex_;
    std::mutex consumer_mutex_;
    
    // 运行状态
    std::atomic<bool> running_;
    
    // 消费线程
    std::thread consuming_thread_;
    
    // 业务层上报消息的回调对象
    kafka_handler notify_handler_;
    
    // Kafka服务器地址
    std::string brokers_;
};

#endif // KAFKA_HPP